package com.spark.gaea.mq.rabbit.listener;

import com.alibaba.druid.pool.DruidDataSource;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.spark.gaea.utils.json.JsonUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpRejectAndDontRequeueException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
import org.springframework.amqp.support.converter.DefaultClassMapper;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.ErrorHandler;

import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.UUID;

@Component
public class WriteToDatabaseErrorHandler implements ErrorHandler {
	private static final Logger LOG;
	@Autowired
	private DruidDataSource dataSource;
	private DefaultMessageConverter messageConverter;
	private ObjectMapper objectMapper;
	private JdbcTemplate jdbcTemplate;

	public WriteToDatabaseErrorHandler() {
		this.messageConverter = new DefaultMessageConverter();
		this.objectMapper = JsonUtil.getMapper();
	}

	@PostConstruct
	public void init() {
		this.jdbcTemplate = new JdbcTemplate(this.dataSource);
	}

	@Override
	public void handleError(final Throwable t) {
		WriteToDatabaseErrorHandler.LOG.warn("Execution of Rabbit message listener failed.", t);
		if (this.causeChainContainsARADRE(t)) {
			throw new AmqpRejectAndDontRequeueException("Error Handler converted exception to fatal", t);
		}
		final ListenerExecutionFailedException lefe = (ListenerExecutionFailedException) t;
		final Message failedMessage = lefe.getFailedMessage();
		if (this.writeToDatabase(failedMessage)) {
			throw new AmqpRejectAndDontRequeueException("Write failed message to database", t);
		}
	}

	private boolean writeToDatabase(final Message message) {
		if (this.jdbcTemplate == null) {
			return false;
		}
		try {
			final int i = this.jdbcTemplate
					.update("INSERT INTO mq_failed_message (id, body, properties) VALUES (?, ?, ?)",
							UUID.randomUUID().toString(),
							this.objectMapper.writeValueAsString(this.messageConverter.fromMessage(message)),
							this.objectMapper.writeValueAsString(message.getMessageProperties()));
			return i > 0;
		} catch (JsonProcessingException e) {
			WriteToDatabaseErrorHandler.LOG.warn("Converter JSON failed.", e);
			return false;
		}
	}

	private boolean causeChainContainsARADRE(final Throwable t) {
		for (Throwable cause = t.getCause(); cause != null; cause = cause.getCause()) {
			if (cause instanceof AmqpRejectAndDontRequeueException) {
				return true;
			}
		}
		return false;
	}

	public DefaultMessageConverter getMessageConverter() {
		return this.messageConverter;
	}

	public void setMessageConverter(final DefaultMessageConverter messageConverter) {
		this.messageConverter = messageConverter;
	}

	static {
		LOG = LoggerFactory.getLogger(WriteToDatabaseErrorHandler.class);
	}

	private class DefaultMessageConverter extends Jackson2JsonMessageConverter {
		private DefaultClassMapper defaultClassMapper;

		public DefaultMessageConverter() {
			this.initializeClassMapper();
		}

		protected void initializeClassMapper() {
			(this.defaultClassMapper = new DefaultClassMapper()).setDefaultType(HashMap.class);
			this.defaultClassMapper.setDefaultHashtableClass(HashMap.class);
			this.setClassMapper(this.defaultClassMapper);
		}
	}
}
